package com.atguigu.gmall.realtime.join;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/4/23 15:20
 */
public class LookupJoin extends BaseSQLApp {
    public static void main(String[] args) {
        new LookupJoin().init(
            50005,
            2,
            "LookupJoin"
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        tEnv.executeSql("create table test(" +
                            "id string, " +
                            "pt as proctime() " +  // 使用 lookup join, 则事实表必须提供一个处理时间
                            ")" + SQLUtil.getKafkaSourceDDL("test", "LookupJoin", "csv"));
        
        // 读取字典表数据
        tEnv.executeSql("create table base_dic(" +
                            " dic_code string, " +
                            " dic_name string " +
                            ")with(" +
                            " 'connector' = 'jdbc'," +
                            " 'url' = 'jdbc:mysql://hadoop162:3306/gmall2023?useSSL=false'," +
                            " 'table-name' = 'base_dic', " +
                            " 'lookup.cache' = 'PARTIAL'," +
                            " 'lookup.partial-cache.max-rows' = '200'," +
//                            " 'lookup.partial-cache.expire-after-write' = '20 second', " +
                            " 'lookup.partial-cache.expire-after-access' = '20 second', " +
                            " 'username' = 'root', " +
                            " 'password' = 'aaaaaa' " +
                            ")");
        
        
        Table result = tEnv.sqlQuery("select " +
                                         " test.id, " +
                                         " dic.dic_name " +
                                         "from test " +
                                         "join base_dic for system_time as of test.pt as dic " +
                                         "on test.id=dic.dic_code");
        result.execute().print();
        
    }
}
